package com.fengye.middleware.rabbitmq.boot.ack.manual;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fengye.middleware.rabbitmq.boot.ack.KnowledgeInfo;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 *
 * @author sky
 * @date 2021年06月09日 11:22 上午
 */
@Component
@Slf4j
public class KnowledgeManualConsumer implements ChannelAwareMessageListener {

    @Autowired
    private ObjectMapper objectMapper;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        // 获取消息属性
        MessageProperties messageProperties = message.getMessageProperties();

        // 获取消息分发时的全局唯一标识
        long deliveryTag = messageProperties.getDeliveryTag();
        try {
            // 获取消息体
            byte[] body = message.getBody();

            // 解析消息体
            KnowledgeInfo info = objectMapper.readValue(body, KnowledgeInfo.class);

            log.info("基于 MANUAL 机制：消费者-接收消息：{}", info);

            //执行完业务逻辑后，手动进行确认消费
            // 第一个参数为：消息的分发标志（全局唯一）
            // 第二个参数为：是否允许批量确认消费
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            log.info("基于 MANUAL 机制：消费者-接受失败", e.fillInStackTrace());

            // 如果在处理消息的过程中发生了异常，则依旧需要人为手动确认消费掉该消息
            // 否则该消息将一直留在队列里，从而导致消息的重复消费
            channel.basicReject(deliveryTag, false);
        }

    }

}
